package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.bw.MyKafkaUtil;
import com.bw.bean.OrderCargo;
import com.bw.bean.OrderInfo;
import com.bw.bean.OrderInfoCargoWide;
import com.bw.bean.Order_info_stats;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Iterator;
import java.util.TimeZone;

public class Test6 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(new MyKafkaUtil().getKafkaConsumer("tms_ods_yk9", "test24"));

        SingleOutputStreamOperator<String> OrderCargoDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));


        // 添加水位线
        SingleOutputStreamOperator<OrderInfo> OrderInfoDS = orderInfoDS.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String s) throws Exception {
                // 把JSON转成实体对象
                OrderInfo orderInfo = JSON.parseObject(JSON.parseObject(s).getString("data"), OrderInfo.class);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                long ts = sdf.parse(orderInfo.getCreate_time()).getTime();
                orderInfo.setCreate_ts(ts);
                return orderInfo;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                // 创建时间的时间戳
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo OrderInfo, long l) {
                        return OrderInfo.getCreate_ts();
                    }
                }));

        SingleOutputStreamOperator<OrderCargo> OrderBoundDS = OrderCargoDS.map(new MapFunction<String, OrderCargo>() {
            @Override
            public OrderCargo map(String s) throws Exception {
                OrderCargo OrderCargo = JSON.parseObject(JSON.parseObject(s).getString("data"), OrderCargo.class);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                long ts = sdf.parse(OrderCargo.getCreate_time()).getTime();
                OrderCargo.setCreate_ts(ts);
                return OrderCargo;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderCargo>forBoundedOutOfOrderness(Duration.ofSeconds(10)).withTimestampAssigner(new SerializableTimestampAssigner<OrderCargo>() {
            @Override
            public long extractTimestamp(OrderCargo OrderCargo, long l) {
                return OrderCargo.getCreate_ts();
            }
        }));

        // 做关联

        SingleOutputStreamOperator<OrderInfoCargoWide> joinDS = OrderInfoDS.keyBy(OrderInfo::getId)
                .intervalJoin(OrderBoundDS.keyBy(OrderCargo::getOrder_id))
                .between(Time.minutes(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderCargo, OrderInfoCargoWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderCargo orderCargo, Context context, Collector<OrderInfoCargoWide> collector) throws Exception {
                        collector.collect(new OrderInfoCargoWide(orderInfo, orderCargo));
                    }
                });


        // 分组开窗聚合

        SingleOutputStreamOperator<Order_info_stats> mapDS2 = joinDS.map(new MapFunction<OrderInfoCargoWide, Order_info_stats>() {
            @Override
            public Order_info_stats map(OrderInfoCargoWide orderInfoCargoWide) throws Exception {
                Order_info_stats order_info_stats=new Order_info_stats();
                //处理时区问题
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                long time = sdf.parse(orderInfoCargoWide.getOrderInfo().getCreate_time()).getTime();
//                order_info.setCreate_time(sdf.format(sdf.parse(order_info.getCreate_time())));

                // 截取前10位
                order_info_stats.setCreate_time_by_day(orderInfoCargoWide.getOrderInfo().getCreate_time().substring(0,10));
                // 货物类型
                order_info_stats.setCargo_type(orderInfoCargoWide.getOrderCargo().getCargo_type());
                // 每条当做1个下单
                order_info_stats.setOrder_order_num(1L);
                // 每条数据的金额
                order_info_stats.setOrder_amount(orderInfoCargoWide.getOrderInfo().getAmount());
                order_info_stats.setTs(time);
                return order_info_stats;
            }
        });

        // 用货物类型和当前的天
        // 2024-08-16  10     100
        // 2024-08-16  10      200

        // 2024-08-17  10      300
//        KeyedStream<Order_info_stats, Tuple2<String,String>> keyedStream = mapDS2.keyBy(new KeySelector<Order_info_stats, Tuple2<String,String>>() {
//            @Override
//            public Tuple2<String,String> getKey(Order_info_stats value) throws Exception {
//                return new Tuple2<>(value.getCargo_type(),
//                        value.getCreate_time_by_day());
//            }
//        });
//        keyedStream.print("keyedStream>>");
        // 开窗
//        SingleOutputStreamOperator<Order_info_stats> reduceStream = keyedStream.reduce(new ReduceFunction<Order_info_stats>() {
//            @Override
//            public Order_info_stats reduce(Order_info_stats t1, Order_info_stats t2) throws Exception {
//                t1.setOrder_order_num(t1.getOrder_order_num() + t2.getOrder_order_num());
//                t1.setOrder_amount(t1.getOrder_amount().add(t2.getOrder_amount()));
//                return t1;
//            }
//        });



        // 开窗的写法
        KeyedStream<Order_info_stats, String> keyedStream = mapDS2.keyBy(new KeySelector<Order_info_stats, String>() {
            @Override
            public String getKey(Order_info_stats value) throws Exception {
                // 按照货物类型key
                return value.getCargo_type();
            }
        });

        SingleOutputStreamOperator<Order_info_stats> reduceDS=keyedStream
                .window(SlidingEventTimeWindows.of(Time.days(1),Time.seconds(10)))
                .reduce(new ReduceFunction<Order_info_stats>() {
                    @Override
                    public Order_info_stats reduce(Order_info_stats stats1, Order_info_stats stats2) throws Exception {
//                        stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
//                        stats1.setOrder_order_num(stats1.getOrder_order_num() + stats2.getOrder_order_num());
                        Order_info_stats order_info_stats = new Order_info_stats();
                        order_info_stats.setOrder_order_num(stats1.getOrder_order_num()+stats2.getOrder_order_num());
                        order_info_stats.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                        order_info_stats.setCargo_type(stats1.getCargo_type());
                        order_info_stats.setCargo_type_name(stats1.getCargo_type_name());
                        return order_info_stats;
                    }
                }, new WindowFunction<Order_info_stats, Order_info_stats,  String, TimeWindow>() {
                    @Override
                    public void apply(String stringStringTuple2, TimeWindow timeWindow, Iterable<Order_info_stats> iterable, Collector<Order_info_stats> collector) throws Exception {
                        //取出窗口开始和结束时间
                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String stt = sdf.format(start);
                        String edt = sdf.format(end);

                        //将窗口时间设置进JavaBean对象中

                        Iterator<Order_info_stats> iterator = iterable.iterator();
                        while (iterator.hasNext()){
                            Order_info_stats Order_info_stats = iterator.next();
                            Order_info_stats.setStt(stt);
                            Order_info_stats.setEdt(edt);
                            //写入数据
                            collector.collect(Order_info_stats);
                        }

                    }
                });

        reduceDS.print("reduceStream>>");
        env.execute();
    }
}
